package org.funtester.performance.books.chapter09.section3;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.serialization.StringSerializer;
import org.funtester.performance.books.chapter03.common.ThreadTool;
import org.funtester.performance.books.chapter03.section3.ThreadTask;
import org.funtester.performance.books.chapter03.section4.TaskExecutor;
import org.funtester.performance.books.chapter04.section3.RandomTool;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * 生产者性能测试用例
 */
public class ProduceCase {

    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();// 创建配置对象
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");// 指定kafka集群地址
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 指定key序列化器
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());// 指定value序列化器
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, "3");// 重试次数
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");// 0-不需要确认,1-需要leader确认,all-需要leader和follower确认
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "128");// 批次大小
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");// 缓冲区大小
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "100");// 等待时间,单位毫秒,默认是0,表示立即发送,设置成10表示10毫秒后发送
        properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, CompressionType.LZ4.name);// 压缩类型
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);// 创建生产者对象
        String topic = "user_behavior";// 指定topic
        int totalNum = 1000;// 总执行次数
        int threadNum = 8;// 线程数
        int rumpUpTime = 20;// Rump-Up运行时间
        int interval = 1;// 发送消息间隔时间,单位毫秒
        List<ThreadTask> tasks = new ArrayList<>();// 任务集合
        List<Integer> userIds = new ArrayList<>();// 用户ID集合
        for (int i = 100000; i < 2000000; i++) {// 生成100000-2000000的用户ID
            userIds.add(i);// 添加到用户ID集合
        }
        for (int i = 0; i < threadNum; i++) {
            ThreadTask threadTask = new ThreadTask() {

                @Override
                public void test() {
                    int userId = RandomTool.getRandomElement(userIds);// 随机获取用户ID
                    UserBehavior userBehavior = new UserBehavior(userId, "浏览商品", "商品信息" + RandomTool.getRandomString(100), System.currentTimeMillis());// 浏览商品行为消息
                    producer.send(new ProducerRecord<>(topic, String.valueOf(userId), userBehavior.toString()), (recordMetadata, e) -> {
                        if (e != null) {// 发送失败
                            // 处理失败逻辑,此处省略
                        } else {
                            // 处理成功逻辑,此处省略
                        }
                    });// 发送消息
                    ThreadTool.sleep(interval);// 线程休眠,防止发送消息过快
                }

            };
            threadTask.totalNum = totalNum;// 设置执行次数
            threadTask.costTime = new ArrayList<>(totalNum);// 设置任务的执行时间集合,设置容量,避免频繁扩容
            tasks.add(threadTask);// 添加任务到任务集合
        }
        TaskExecutor taskExecutor = new TaskExecutor(tasks, "用户行为买点信息生产者", rumpUpTime);
        taskExecutor.start();// 启动任务执行器
        producer.close();// 关闭生产者
    }


}
